package com.jwater.core.net.group;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.jwater.core.common.S;
import com.jwater.core.net.NetException;
import com.jwater.core.net.Node;
import com.jwater.core.net.NodeHandler;
import com.jwater.core.net.NodeHandlerAdapter;
import com.jwater.core.net.SimpleNode;
import com.jwater.core.net.group.data.DurableData;
import com.jwater.core.net.group.data.DurableDataItem;
import com.jwater.core.net.group.data.DurableDataManager;
import com.jwater.core.net.group.lock.GroupLock;
import com.jwater.core.serialize.SerializeUtils;

/**
 * @description:
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class SimpleGroupNode implements GroupNode {
    protected static Logger logger = LoggerFactory.getLogger(SimpleGroupNode.class);

    public static final String TYPE_SEND_MEMBER_ID = "$sendMemberId";
    public static final String TYPE_SET_SHARE_DATA = "$setShareData";
    public static final String TYPE_REMOVE_SHARE_DATA = "$removeShareData";
    public static final String TYPE_REQUEST_SHARE_DATA = "$requestShareData";
    public static final String TYPE_LOCK = "$lock";
    public static final String TYPE_UNLOCK = "$unlock";

    private Map<Integer, Member> memberMap;
    private Member me;
    private GroupHandler groupHandler;
    private NodeHandler nodeHandler;
    private Map<Integer, Node> memberNodes = new ConcurrentHashMap<>();
    private DurableDataManager dataManager = new DurableDataManager();
    private GroupLock groupLock = new GroupLock();
    private volatile boolean needRequestShareData = true;

    public SimpleGroupNode(List<Member> members) {
        memberMap = new HashMap<>(members.size());
        for (Member member : members) {
            memberMap.put(member.getMemberId(), member);
        }
        dataManager.load();
        dataManager.startDurableThread();
    }

    @Override
    public void setGroupHandler(GroupHandler handler) {
        this.groupHandler = handler;
    }

    @Override
    public void setNodeHandler(NodeHandler handler) {
        this.nodeHandler = handler;
    }

    @Override
    public void joinGroup(int memberId) {
        me = memberMap.get(memberId);
        if (me == null) {
            throw new NetException("member can not find for:" + memberId);
        }
        MemberHandler memberHandler = new MemberHandler();
        Node serverNode = new SimpleNode();
        serverNode.setHandler(memberHandler);
        serverNode.startAsServer(me.getInnerPort());
        for (Member member : memberMap.values()) {
            if (member.getMemberId() < me.getMemberId()) {
                Node client = new SimpleNode();
                client.setAttribute("meId", me.getMemberId());
                client.setAttribute("memberId", member.getMemberId());
                client.setHandler(memberHandler);
                GroupExecutorUtils.execute(() -> {
                    client.startAsClient(member.getIp(), member.getInnerPort(), true);
                });
            }
        }
        startOuterServer();
    }

    protected void startOuterServer() {
        int waitTime = 3000;
        int time = 0;
        while (memberMap.size() != memberNodes.size() && time < waitTime) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                logger.error(e.toString(), e);
            }
            time += 100;
        }
        Node serverNode = new SimpleNode();
        serverNode.setHandler(nodeHandler);
        serverNode.startAsServer(me.getOuterPort());
    }

    @Override
    public void sendDataToMembers(String dataType, byte[] data) {
        CountDownLatch latch = new CountDownLatch(memberMap.size());
        for (Integer memberId : memberMap.keySet()) {
            Node node = memberNodes.get(memberId);
            GroupExecutorUtils.execute(() -> {
                sendDataToMember(dataType, data, node, latch);
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error(e.toString(), e);
        }
    }

    @Override
    public Map<Integer, byte[]> callMembers(String method, byte[] data) {
        Map<Integer, byte[]> result = new HashMap<>();
        CountDownLatch latch = new CountDownLatch(memberMap.size());
        for (Integer memberId : memberMap.keySet()) {
            Node node = memberNodes.get(memberId);
            GroupExecutorUtils.execute(() -> {
                try {
                    if (node != null) {
                        result.put(memberId, node.syncCall(method, data));
                    }
                } catch (Exception e) {
                    logger.error(e.toString(), e);
                } finally {
                    latch.countDown();
                }
            });
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            logger.error(e.toString(), e);
        }
        return result;
    }

    private void sendDataToMember(String dataType, byte[] data, Node node, CountDownLatch latch) {
        try {
            if (node != null) {
                node.sendData(dataType, data);
            }
        } catch (Exception e) {
            logger.error(e.toString(), e);
        } finally {
            latch.countDown();
        }
    }

    @Override
    public void lock(String key) {
        // 分布式锁，以编号最小的节点作为主控节点
        int min = minMemberId();
        if (min == me.getMemberId()) {
            // 主控节点就是自己，直接本地锁
            groupLock.lock(key, min, Thread.currentThread().getId());
        } else {
            Node node = memberNodes.get(min);
            if (node == null) {
                lock(key);
            } else {
                try {
                    node.syncCall(TYPE_LOCK, SerializeUtils.serializeArray(key, min, Thread.currentThread().getId()));
                } catch (Exception e) {
                    logger.error(e.toString(), e);
                }
            }
        }
    }

    @Override
    public void unlock(String key) {
        int min = minMemberId();
        if (min == me.getMemberId()) {
            groupLock.unlock(key);
        } else {
            Node node = memberNodes.get(min);
            if (node != null) {
                try {
                    node.syncCall(TYPE_UNLOCK, key.getBytes(S.charset));
                } catch (Exception e) {
                    logger.error(e.toString(), e);
                }
            }
        }
    }

    @Override
    public void storeDataItem(String key, String itemKey, DurableDataItem data) {
        byte[] bytes = SerializeUtils.serializeArray(key, itemKey, data.getVersion(), data.getData(), true);
        dataManager.setItem(key, itemKey, data.getVersion(), data.getData());
        GroupExecutorUtils.execute(() -> {
            DataItemEvent event = new DataItemEvent();
            event.setType(DataItemEvent.TYPE_STORE);
            event.setKey(key);
            event.setItemKey(itemKey);
            event.setVersion(data.getVersion());
            event.setData(data.getData());
            groupHandler.onDataItemEvent(event);
        });
        sendDataToMembers(TYPE_SET_SHARE_DATA, bytes);
    }

    @Override
    public DurableDataItem getDataItem(String key, String itemKey) {
        return dataManager.getItem(key, itemKey);
    }

    @Override
    public void removeDataItem(String key, String itemKey) {
        dataManager.removeItem(key, itemKey);
        byte[] bytes = SerializeUtils.serializeArray(key, itemKey, true);
        GroupExecutorUtils.execute(() -> {
            DataItemEvent event = new DataItemEvent();
            event.setType(DataItemEvent.TYPE_REMOVE);
            event.setKey(key);
            event.setItemKey(itemKey);
            groupHandler.onDataItemEvent(event);
        });
        sendDataToMembers(TYPE_REMOVE_SHARE_DATA, bytes);
    }

    @Override
    public void setData(String key, boolean isDurable) {
        dataManager.set(key, isDurable);
    }

    @Override
    public Map<String, DurableDataItem> getItemsByPrefix(String key, String prefix) {
        return dataManager.getItemsByPrefix(key, prefix);
    }

    @Override
    public Map<String, DurableDataItem> getItems(String key) {
        DurableData durableData = dataManager.getData(key);
        if (durableData == null) {
            return new HashMap<String, DurableDataItem>();
        }
        return durableData.items();
    }

    protected int minMemberId() {
        int min = me.getMemberId();
        for (Integer memberId : memberNodes.keySet()) {
            if (memberId < min) {
                min = memberId;
            }
        }
        return min;
    }

    /**
     * 是否是主控节点
     *
     * @return
     */
    @Override
    public boolean isLeader() {
        int min = minMemberId();
        if (min == me.getMemberId()) {
            return true;
        }
        return false;
    }

    private class MemberHandler extends NodeHandlerAdapter {

        @Override
        public void onConnect(Node node) {
            sendMemberId(node);
            Integer memberId = (Integer) node.getAttribute("memberId");
            if (memberId != null) {
                closeOldNode(node, memberId);
                memberNodes.put(memberId, node);
                if (groupHandler != null) {
                    GroupExecutorUtils.execute(() -> {
                        groupHandler.onJoin(memberId, node);
                    });
                }
            }
            if (needRequestShareData) {
                needRequestShareData = false;
                try {
                    node.sendData(TYPE_REQUEST_SHARE_DATA, new byte[0]);
                } catch (Exception e) {
                    logger.error(e.toString(), e);
                }
            }
        }

        private void sendMemberId(Node node) {
            Integer meId = (Integer) node.getAttribute("meId");
            if (meId != null) {
                try {
                    node.sendData(TYPE_SEND_MEMBER_ID, String.valueOf(meId).getBytes(S.charset));
                } catch (Exception e) {
                    logger.error(e.toString(), e);
                }
            }
        }

        private void closeOldNode(Node node, Integer memberId) {
            Node oldNode = memberNodes.get(memberId);
            if (oldNode != null && node != oldNode) {
                oldNode.close();
            }
        }

        @Override
        public void onDisconnect(Node node) {
            Integer memberId = (Integer) node.getAttribute("memberId");
            if (memberId != null) {
                closeOldNode(node, memberId);
                memberNodes.remove(memberId);
                if (groupHandler != null) {
                    GroupExecutorUtils.execute(() -> {
                        groupHandler.onLeave(memberId, node);
                    });
                }
            }
            if (memberNodes.isEmpty()) {
                needRequestShareData = true;
            }
        }

        @Override
        public byte[] onSyncCall(Node node, String method, byte[] args) {
            if (TYPE_LOCK.equals(method)) {
                Object[] objs = SerializeUtils.deserializeArray(args);
                String key = (String) objs[0];
                int memberId = (int) objs[1];
                int threadId = (int) objs[2];
                groupLock.lock(key, memberId, threadId);
            } else if (TYPE_UNLOCK.equals(method)) {
                String key = S.newString(args);
                groupLock.unlock(key);
            } else {
                Integer memberId = (Integer) node.getAttribute("memberId");
                if (groupHandler != null && memberId != null) {
                    return groupHandler.onMemberCall(memberId, node, method, args);
                }
            }
            return null;
        }

        @Override
        public void onData(Node node, String dataType, byte[] data) {
            if (TYPE_SEND_MEMBER_ID.equals(dataType)) {
                setMemberId(node, data);
            } else if (TYPE_SET_SHARE_DATA.equals(dataType)) {
                setShareData(data);
            } else if (TYPE_REMOVE_SHARE_DATA.equals(dataType)) {
                removeShareData(data);
            } else if (TYPE_REQUEST_SHARE_DATA.equals(dataType)) {
                syncShareData(node);
            } else {
                doOnData(node, dataType, data);
            }
        }

        protected void doOnData(Node node, String dataType, byte[] data) {
            Integer memberId = (Integer) node.getAttribute("memberId");
            if (groupHandler != null && memberId != null) {
                GroupExecutorUtils.execute(() -> {
                    groupHandler.onMemberData(memberId, node, dataType, data);
                });
            }
        }

        protected void setMemberId(Node node, byte[] data) {
            try {
                int memberId = Integer.parseInt(new String(data, 0, data.length, S.charset));
                node.setAttribute("memberId", memberId);
                memberNodes.put(memberId, node);
                if (groupHandler != null) {
                    GroupExecutorUtils.execute(() -> {
                        groupHandler.onJoin(memberId, node);
                    });
                }
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
        }

        protected void setShareData(byte[] data) {
            Object[] objs = SerializeUtils.deserializeArray(data);
            String key = (String) objs[0];
            String itemKey = (String) objs[1];
            long version = (long) objs[2];
            String d = (String) objs[3];
            boolean notify = (boolean) objs[4];
            dataManager.setItem(key, itemKey, version, d);
            if (notify) {
                GroupExecutorUtils.execute(() -> {
                    DataItemEvent event = new DataItemEvent();
                    event.setType(DataItemEvent.TYPE_STORE);
                    event.setKey(key);
                    event.setItemKey(itemKey);
                    event.setVersion(version);
                    event.setData(d);
                    groupHandler.onDataItemEvent(event);
                });
            }
        }

        protected void removeShareData(byte[] data) {
            Object[] objs = SerializeUtils.deserializeArray(data);
            String key = (String) objs[0];
            String itemKey = (String) objs[1];
            boolean notify = (boolean) objs[2];
            dataManager.removeItem(key, itemKey);
            if (notify) {
                GroupExecutorUtils.execute(() -> {
                    DataItemEvent event = new DataItemEvent();
                    event.setType(DataItemEvent.TYPE_REMOVE);
                    event.setKey(key);
                    event.setItemKey(itemKey);
                    groupHandler.onDataItemEvent(event);
                });
            }
        }

        protected void syncShareData(Node node) {
            for (Entry<String, DurableData> en1 : dataManager.datas().entrySet()) {
                for (Entry<String, DurableDataItem> en2 : en1.getValue().items().entrySet()) {
                    byte[] bytes = SerializeUtils.serializeArray(en1.getKey(), en2.getKey(),
                            en2.getValue().getVersion(), en2.getValue().getData(), false);
                    try {
                        node.sendData(TYPE_SET_SHARE_DATA, bytes);
                    } catch (Exception e) {
                        logger.error(e.toString(), e);
                    }
                }
            }
        }
    }
}
